package com.atbeijing.bigdata.spark.core.acc

import org.apache.spark.util.AccumulatorV2
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable

object Spark03_Var_Acc {

    def main(args: Array[String]): Unit = {
        val conf = new SparkConf().setMaster("local").setAppName("Acc")
        val sc = new SparkContext(conf)
        val rdd = sc.makeRDD(List(
            ("Hello", 1), ("Hello", 2), ("Hello", 3), ("spark", 1), ("spark", 2), ("scala", 1)
        ),2)
        //声明累加器
        val wordCount = new WordCountAcc()
        //注册累加器
        sc.register(wordCount, "WordCountAcc")
        //没有shuffle阶段所以速度提升
        rdd.foreach (
            t => {
                //使用累加器
                wordCount.add(t)
            }
        )
        //获取累加器的值
        println(wordCount.value)//Map(spark -> 3, scala -> 1, Hello -> 6)
        sc.stop()
    }
    // 自定义WordCount累加器
    // 1. 继承AccumulatorV2
    // 2. 确定泛型
    //    IN : 向累加器中增加的值的类型
    //    OUT : 累加器的结果就是输出
    // 3. 重写方法
    class WordCountAcc extends AccumulatorV2[(String, Int), mutable.Map[String, Int]]{

        //定义累加器使用什么容器 =>map
        private val wordCountMap = mutable.Map[String, Int]()

        //判断累加器是否为初始状态
        override def isZero: Boolean = wordCountMap.isEmpty

        //复制一个新的累加器
        override def copy(): AccumulatorV2[(String, Int), mutable.Map[String, Int]] =  {
            new WordCountAcc()
        }

        //重置累加器
        override def reset(): Unit = wordCountMap.clear()

        // 将数据向累加器中增加,也就是将传来的一个个tuple合并到wordCountMap
        override def add(v: (String, Int)): Unit = {
            val (word, cnt) = v
            val oldCnt = wordCountMap.getOrElse(word, 0)
            wordCountMap.update(word, cnt + oldCnt)
        }

        // 合并累加器
        override def merge(other: AccumulatorV2[(String, Int), mutable.Map[String, Int]]): Unit =  {
            //参数累加器
            val otherMap: mutable.Map[String, Int] = other.value
            //遍历累参数累加器中的数据合并到调用merge方法的对象的wordCountMap中
            otherMap.foreach {
                case ( word, count ) => {
                    val oldCnt = this.wordCountMap.getOrElse(word, 0)
                    this.wordCountMap.update(word, count + oldCnt)
                }
            }
        }

        //累加器的结果
        override def value: mutable.Map[String, Int] = wordCountMap
    }
}
